package com.atguigu.edu.function;


import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;

import com.atguigu.edu.util.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;

import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;


public abstract class AsyncDimFunction<T> extends RichAsyncFunction<T, T> {
    
    public abstract String getTable();
    public abstract String getId(T input);
    
    public abstract void addDim(T input, JSONObject dim);
    
    
    private ThreadPoolExecutor threadPool;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        // 获取线程池
        threadPool = ThreadPoolUtil.getTreadPool();
        
    }
    
    @Override
    public void asyncInvoke(T input,
                            ResultFuture<T> resultFuture) {

        threadPool.submit(new Runnable() {
            @Override
            public void run() {
                // 每个线程分别获取一个 redis 的客户端和 phoenix 连接
                try {
                    Jedis redisClient = RedisUtil.getRedisClient();
                    DruidPooledConnection phoenixConn = DruidDSUtil.getPhoenixConn();
                    JSONObject dim = DimUtil.readDim(redisClient, phoenixConn, getTable(), getId(input));
                    addDim(input, dim);
                    resultFuture.complete(Collections.singletonList(input));

                    redisClient.close();
                    JdbcUtil.closeConnection(phoenixConn);
                    
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }
        });
    }
}
